package com.jc.demo.hadoop.jdata;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.orc.TypeDescription;
import org.apache.orc.mapred.OrcStruct;

import java.io.IOException;

/***
 CREATE TABLE `t_test_orc_new`(
 `phone` string,
 `firsDate` string,
 `lastDate` string)
 ROW FORMAT SERDE
 'org.apache.hadoop.hive.ql.io.orc.OrcSerde'
 STORED AS INPUTFORMAT
 'org.apache.hadoop.hive.ql.io.orc.OrcInputFormat'
 OUTPUTFORMAT
 'org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat'
 LOCATION
 'hdfs://s1.bigdata.ucpaas.com:9000/user/testorc3'


 */
public class BlankPhoneReducer extends
        Reducer<Text, Text, NullWritable, OrcStruct> {
    private TypeDescription schema = TypeDescription
            .fromString("struct<phone:string,firsDate:string,lastDate:string>");
    private OrcStruct pair = (OrcStruct) OrcStruct.createValue(schema);

    private final NullWritable nw = NullWritable.get();

    public void reduce(Text key, Iterable<Text> values, Context output)
            throws IOException, InterruptedException {


        String max = null;
        String min = null;
        for (Text date : values) {
            if (max == null)
                max = date.toString();
            if (min == null)
                min = date.toString();

            if (date.toString().compareTo(max) > 0)
                max = date.toString();
            if (date.toString().compareTo(min) < 0)
                min = date.toString();
        }

        pair.setFieldValue(0, key);
        pair.setFieldValue(1, new Text(min));
        pair.setFieldValue(2, new Text(max));
        output.write(nw, pair);


    }
}
